package com.gxlevi.task

import com.gxlevi.bean.ClickLogWide
import com.gxlevi.util.HBaseUtil
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, WindowedStream}
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.api.scala._

case class ChannelArea(
                        var channelId: String,
                        var area: String,
                        var date: String,
                        var pv: Long,
                        var uv: Long,
                        var newCount: Long,
                        var oldCount: Long
                      )

object ChannelAreaTask extends BaseTask[ChannelArea] {
  override def map(clickWideDataStream: DataStream[ClickLogWide]): DataStream[ChannelArea] = {
    clickWideDataStream.flatMap {
      clickLogWide =>
        List(
          ChannelArea(
            clickLogWide.channelID,
            clickLogWide.address,
            clickLogWide.yearMonth,
            clickLogWide.count,
            clickLogWide.isMonthNew,
            clickLogWide.isNew,
            isOld(clickLogWide.isNew, clickLogWide.isMonthNew)
          ),
          ChannelArea(
            clickLogWide.channelID,
            clickLogWide.address,
            clickLogWide.yearMonthDay,
            clickLogWide.count,
            clickLogWide.isDayNew,
            clickLogWide.isNew,
            isOld(clickLogWide.isNew, clickLogWide.isDayNew)
          ),
          ChannelArea(
            clickLogWide.channelID,
            clickLogWide.address,
            clickLogWide.yearMonthDayHour,
            clickLogWide.count,
            clickLogWide.isHourNew,
            clickLogWide.isNew,
            isOld(clickLogWide.isNew, clickLogWide.isHourNew)
          )
        )
    }
  }

  override def keyBy(mapStream: DataStream[ChannelArea]): KeyedStream[ChannelArea, String] = {
    mapStream.keyBy {
      area => (area.channelId + ":" + area.area + ":" + area.date)
    }
  }

  override def reduce(windowedStream: WindowedStream[ChannelArea, String, TimeWindow]): DataStream[ChannelArea] = {
    windowedStream.reduce {
      (t1, t2) =>
        ChannelArea(t1.channelId, t1.area, t1.date, t1.pv + t2.pv, t1.uv + t2.uv, t1.newCount + t2.newCount, t1.oldCount + t2.oldCount)
    }
  }

  override def sinkToHBase(reduceDataStream: DataStream[ChannelArea]): Unit = {
    reduceDataStream.addSink {
      area => {
        tableName = "channel_area"
        rowKey = area.channelId + ":" + area.area + ":" + area.date
        val areaColumn = "area"

        val map: Map[String, String] = HBaseUtil.getMapData(tableName, rowKey, clfName, List(pvColumn, uvColumn, newCountColumn, oldCountColumn))
        val totalPv = getTotal(map, pvColumn, area.pv)
        val totalUv = getTotal(map, uvColumn, area.uv)
        val totalNewCount = getTotal(map, newCountColumn, area.newCount)
        val totalOldCount = getTotal(map, oldCountColumn, area.oldCount)

        HBaseUtil.putMapData(tableName, rowKey, clfName, Map(
          channelIdColumn -> area.channelId,
          areaColumn -> area.area,
          dateColumn -> area.date,
          pvColumn -> totalPv,
          uvColumn -> totalUv,
          newCountColumn -> totalNewCount,
          oldCountColumn -> totalOldCount
        ))
      }
    }
  }
}
